/*
 * Copyright (C) 2020 The LINN Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 *
 * NOTICE:
 *
 * The SubscriberRegistry Source code from Guava Authors.
 */

package com.juqimiao.linn.event;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Throwables;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.*;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.UncheckedExecutionException;
import com.google.j2objc.annotations.Weak;
import com.google.common.base.Objects;

import java.lang.reflect.Method;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.Arrays;

import com.juqimiao.linn.exception.FrameworkException;
import com.juqimiao.linn.logging.Logger;
import org.checkerframework.checker.nullness.qual.Nullable;

/**
 * 单个事件总线的订阅者的注册。
 *
 * Registry of subscribers to a single event bus.
 *
 * @author Colin Decker
 */
public class SubscriberRegistry {

    private Logger logger = Logger.getLogger(SubscriberRegistry.class);
    /**
     * All registered subscribers, indexed by event type.
     *
     * <p>The {@link CopyOnWriteArraySet} values make it easy and relatively lightweight to get an
     * immutable snapshot of all current subscribers to an event without any locking.
     */
    private final ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers =
            Maps.newConcurrentMap();

    /** The event bus this registry belongs to. */

    @Weak
    private final EventBus bus;

    SubscriberRegistry(EventBus bus) {
        this.bus = checkNotNull(bus);
    }

    /** Registers all subscriber methods on the given listener object. */

    void register(Object listener) {
        logger.info(String.format("Begin register %s ", listener));

        Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

        for (Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {
            Class<?> eventType = entry.getKey();
            Collection<Subscriber> eventMethodsInListener = entry.getValue();

            CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);

            if (eventSubscribers == null) {
                CopyOnWriteArraySet<Subscriber> newSet = new CopyOnWriteArraySet<>();
                eventSubscribers =
                        MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);
            }

            eventSubscribers.addAll(eventMethodsInListener);

            logger.info(String.format("End register %s ", listener));
        }
    }

    void register(EventHandlerI listener) {
        register((Object) listener);
    }

    /** Unregisters all subscribers on the given listener object. */
    void unregister(Object listener) {
        logger.info(String.format("Begin unregister %s ", listener));
        Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

        for (Map.Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {
            Class<?> eventType = entry.getKey();
            Collection<Subscriber> listenerMethodsForType = entry.getValue();

            CopyOnWriteArraySet<Subscriber> currentSubscribers = subscribers.get(eventType);
            if (currentSubscribers == null || !currentSubscribers.removeAll(listenerMethodsForType)) {
                // if removeAll returns true, all we really know is that at least one subscriber was
                // removed... however, barring something very strange we can assume that if at least one
                // subscriber was removed, all subscribers on listener for that event type were... after
                // all, the definition of subscribers on a particular class is totally static
                throw new IllegalArgumentException(
                        "missing event subscriber for an annotated method. Is " + listener + " registered?");
            }

            // don't try to remove the set if it's empty; that can't be done safely without a lock
            // anyway, if the set is empty it'll just be wrapping an array of length 0
            logger.info(String.format("End unregister %s ", listener));
        }
    }

    @VisibleForTesting
    Set<Subscriber> getSubscribersForTesting(Class<?> eventType) {
        return MoreObjects.firstNonNull(subscribers.get(eventType), ImmutableSet.<Subscriber>of());
    }

    /**
     * Gets an iterator representing an immutable snapshot of all subscribers to the given event at
     * the time this method is called.
     */
    Iterator<Subscriber> getSubscribers(Object event) {
        ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass());

        List<Iterator<Subscriber>> subscriberIterators =
                Lists.newArrayListWithCapacity(eventTypes.size());

        for (Class<?> eventType : eventTypes) {
            CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);
            if (eventSubscribers != null) {
                // eager no-copy snapshot
                subscriberIterators.add(eventSubscribers.iterator());
            }
        }

        return Iterators.concat(subscriberIterators.iterator());
    }

    /**
     * A thread-safe cache that contains the mapping from each class to all methods in that class and
     * all super-classes, that are annotated with {@code @Subscribe}. The cache is shared across all
     * instances of this class; this greatly improves performance if multiple EventBus instances are
     * created and objects of the same class are registered on all of them.
     */
    private static final LoadingCache<Class<?>, ImmutableList<Method>> subscriberMethodsCache =
            CacheBuilder.newBuilder()
                    .weakKeys()
                    .build(new CacheLoader<Class<?>, ImmutableList<Method>>() {
                                @Override
                                public ImmutableList<Method> load(Class<?> concreteClass) throws Exception {
                                    return isEventHandleI(concreteClass) ?
                                            getEventIMethods(concreteClass) :
                                            getAnnotatedMethodsNotCached(concreteClass);
                                }
                            });


    private static boolean isEventHandleI(Class<?> concreteClass) {
        Class<?>[] clazz = concreteClass.getInterfaces();
        boolean result = false;
        if (clazz != null && clazz.length > 0) {
            for (Class<?> clz : clazz) {
                if (clz.equals(EventHandlerI.class)) {
                    result = true;
                }
            }
        }
        return result;
    }

    private static ImmutableList<Method> getAnnotatedMethods(Class<?> clazz) {
        return subscriberMethodsCache.getUnchecked(clazz);
    }

    /**
     * Returns all subscribers for the given listener grouped by the type of event they subscribe to.
     */
    private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
        Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
        Class<?> clazz = listener.getClass();
        /** make change for EventhandlerI */
        for (Method method : getAnnotatedMethods(clazz)) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            Class<?> eventType = parameterTypes[0];
            methodsInListener.put(eventType, Subscriber.create(bus, listener, method));
        }
        return methodsInListener;
    }
    /**  */
    private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz) {
        Set<? extends Class<?>> supertypes = TypeToken.of(clazz).getTypes().rawTypes();
        Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
        for (Class<?> supertype : supertypes) {
            for (Method method : supertype.getDeclaredMethods()) {
                if (method.isAnnotationPresent(Subscribe.class) && !method.isSynthetic()) {
                    // TODO: 需要检查泛型参数类型并且给出相应错误。
                    Class<?>[] parameterTypes = method.getParameterTypes();
                    /** Todo: Throw FrameworkException */
                    checkArgument(parameterTypes.length == 1,
                            "Method %s has @Subscribe annotation but has %s parameters."
                                    + "Subscriber methods must have exactly 1 parameter.",
                            method,
                            parameterTypes.length);
                    MethodIdentifier ident = new MethodIdentifier(method);
                    if (!identifiers.containsKey(ident)) {
                        identifiers.put(ident, method);
                    }
                }
            }
        }
        return ImmutableList.copyOf(identifiers.values());
    }

    /**
     * 返回execute方法信息, 并检查方法是否符合定义。
     *
     * @param clazz {@link EventHandlerI}的{@link Class<?>}反射类。
     *
     * @return      代表execute方法的反射类型。
     */
    private static ImmutableList<Method> getEventIMethods(Class<?> clazz) {
        Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
        Method method = null;
        Method[] methods = clazz.getDeclaredMethods();

        try {
            if (methods.length == 0) {
                throw new FrameworkException("execute method in "
                        + method.getDeclaringClass()
                        + "should has one parameter which implement from EventHandlerI");
            }
            method = Arrays.stream(methods)
                    .filter(item -> !item.isBridge()
                            && "execute".equals(item.getName()))
                    .findFirst()
                    .get();
        } catch (Exception e) {
            throw new FrameworkException("execute method in "
                    + method.getDeclaringClass()
                    + "should has one parameter which implement from EventHandlerI");
        }

        if (method == null) {
            throw new FrameworkException("execute method in "
                    + method.getDeclaringClass()
                    + "should has one parameter which implement from EventHandlerI");
        }

        Class<?>[] parameterTypes = method.getParameterTypes();

        checkArgument(parameterTypes.length == 1,
                "Method %s of EventHandlerI but has %s parameters."
                        + "Subscriber methods must have exactly 1 parameter.",
                method,
                parameterTypes.length);

        MethodIdentifier ident = new MethodIdentifier(method);
        if (!identifiers.containsKey(ident)) {
            identifiers.put(ident, method);
        }

        return ImmutableList.copyOf(identifiers.values());
    }

    /** Global cache of classes to their flattened hierarchy of supertypes. */
    private static final LoadingCache<Class<?>, ImmutableSet<Class<?>>> flattenHierarchyCache =
            CacheBuilder.newBuilder()
                    .weakKeys()
                    .build(new CacheLoader<Class<?>, ImmutableSet<Class<?>>>() {
                                // <Class<?>> is actually needed to compile
                                @SuppressWarnings("RedundantTypeArguments")
                                @Override
                                public ImmutableSet<Class<?>> load(Class<?> concreteClass) {
                                    return ImmutableSet.<Class<?>>copyOf(
                                            TypeToken.of(concreteClass).getTypes().rawTypes());
                                }
                            });

    /**
     * Flattens a class's type hierarchy into a set of {@code Class} objects including all
     * superclasses (transitively) and all interfaces implemented by these superclasses.
     */
    @VisibleForTesting
    static ImmutableSet<Class<?>> flattenHierarchy(Class<?> concreteClass) {
        try {
            return flattenHierarchyCache.getUnchecked(concreteClass);
        } catch (UncheckedExecutionException e) {
            Throwables.propagateIfPossible(e.getCause(), FrameworkException.class);
            throw new RuntimeException("unknown", e);
        }
    }

    /**
     * 通过方法名 + 方法参数唯一确定一个方法id。
     */
    private static final class MethodIdentifier {

        private final String name;
        private final List<Class<?>> parameterTypes;

        MethodIdentifier(Method method) {
            this.name = method.getName();
            this.parameterTypes = Arrays.asList(method.getParameterTypes());
        }

        @Override
        public int hashCode() {
            return Objects.hashCode(name, parameterTypes);
        }

        @Override
        public boolean equals(@Nullable Object o) {
            if (o instanceof MethodIdentifier) {
                MethodIdentifier ident = (MethodIdentifier) o;
                return name.equals(ident.name)
                        && parameterTypes.equals(ident.parameterTypes);
            }
            return false;
        }
    }
}
